package net.oschina.j2cache.redis;

import net.oschina.j2cache.*;
import net.oschina.j2cache.util.SerializationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * Redis 缓存基于Hashs实现
 *
 * @author wendal
 */
public class RedisCache implements Cache {

    private final static Logger log = LoggerFactory.getLogger(RedisCache.class);

    // 记录region
    protected String region0;//原始的，不加namespace前缀
    protected String fullRegion;
    protected byte[] byteFullRegion;
    protected RedisCacheProxy redisCacheProxy;

    private final boolean storedInString;

    public RedisCache(String region, RedisCacheProxy redisCacheProxy) {
        if (region == null || region.isEmpty()) {
            region = CacheManager.DEFAULT_REGION; // 缺省region
            storedInString = true;
        } else {
            storedInString = RedisCacheChannel.getInstance().getStringValueRegionSet().contains(region);
        }
        this.region0 = region;
        this.fullRegion = getRegionName(region0);
        this.byteFullRegion = fullRegion.getBytes();
        this.redisCacheProxy = redisCacheProxy;
    }

    /**
     * 在region里增加一个可选的层级,作为命名空间,使结构更加清晰
     * 同时满足小型应用,多个J2Cache共享一个redis database的场景
     *
     * @param region
     * @return
     */
    private String getRegionName(String region) {
        String nameSpace = J2Cache.getConfig().getProperty("redis.namespace", "");
        if (nameSpace != null && !nameSpace.isEmpty()) {
            region = nameSpace + ":" + region;
        }
        return region;
    }

    protected String getRealKey(Object key) {
        if (key instanceof Number)
            return "I:" + key;
        else if (key instanceof String || key instanceof StringBuilder || key instanceof StringBuffer)
            return "S:" + key;
        return "O:" + key;
    }

    protected byte[] getRealKeyBytes(Object key) {
        return getRealKey(key).getBytes();
    }

    protected String getRedisKeyName(String realKey) {
        StringBuilder sb = new StringBuilder()
                .append(fullRegion)
                .append(":")
                .append(realKey);
        return sb.toString();
    }

    protected byte[] getRedisKeyNameBytes(String realKey) {
        return getRedisKeyName(realKey).getBytes();
    }

    public Object get(Object key) throws CacheException {
        if (null == key)
            return null;
        Object obj = null;
        final String realKey = getRealKey(key);
        try {
            byte[] b = storedInString ? redisCacheProxy.get(getRedisKeyNameBytes(realKey)) : redisCacheProxy.hget(byteFullRegion, realKey.getBytes());
            if (b != null) {
                obj = SerializationUtils.deserialize(b);
                log.debug("GET {}({}) - hit", fullRegion, realKey);
            } else {
                log.debug("GET {}({}) - cache miss", fullRegion, realKey);
            }
        } catch (Exception e) {
            log.error("Error occured when get data from redis2 cache", e);
            if (e instanceof IOException || e instanceof NullPointerException)
                evict(key);
        }
        return obj;
    }

    public void put(Object key, Object value) throws CacheException {
        if (key == null)
            return;
        if (value == null)
            evict(key);
        else {
            final String realKey = getRealKey(key);
            log.debug("PUT {}({})", fullRegion, realKey);
            try {
                if (storedInString)
                    redisCacheProxy.set(getRedisKeyNameBytes(realKey), SerializationUtils.serialize(value));
                else
                    redisCacheProxy.hset(byteFullRegion, realKey.getBytes(), SerializationUtils.serialize(value));
            } catch (Exception e) {
                log.error("Error occured when put data to redis2 cache", e);
                //throw new CacheException(e);
            }
        }
    }

    public void update(Object key, Object value) throws CacheException {
        put(key, value);
    }

    public void evict(Object key) throws CacheException {
        if (key == null)
            return;
        final String realKey = getRealKey(key);
        log.debug("DEL {}({})", fullRegion, realKey);
        try {
            if (storedInString)
                redisCacheProxy.del(getRedisKeyNameBytes(realKey));
            else
                redisCacheProxy.hdel(byteFullRegion, realKey.getBytes());
        } catch (Exception e) {
            log.error("Error occured when evict one data from redis2 cache", e);
            //throw new CacheException(e);
        }
    }

    @SuppressWarnings("rawtypes")
    public void evict(List keys) throws CacheException {
        if (keys == null || keys.size() == 0)
            return;
        try {
            int size = keys.size();
            if (storedInString)
                for (int i = 0; i < size; i++)
                    evict(keys.get(i));
            else {
                log.debug("DEL {}({})", fullRegion, keys);
                byte[][] bkeys = new byte[size][];
                for (int i = 0; i < size; i++) {
                    bkeys[i] = getRedisKeyNameBytes(getRealKey(keys.get(i)));
                }
                redisCacheProxy.hdel(byteFullRegion, bkeys);
            }
        } catch (Exception e) {
            log.error("Error occured when evict multi data from redis2 cache", e);
            //throw new CacheException(e);
        }
    }

    public List<String> keys() throws CacheException {
        try {
            if (storedInString)
                return new ArrayList<String>(redisCacheProxy.keys(fullRegion + ":[ISO]:*"));
            else
                return new ArrayList<String>(redisCacheProxy.hkeys(fullRegion));
        } catch (Exception e) {
            log.error("Error occured when call keys on redis2 cache", e);
            return new ArrayList<String>();
            //throw new CacheException(e);
        }
    }

    public void clear() throws CacheException {
        log.debug("CLEAR {}", fullRegion);
        try {
            if (storedInString) {//redis中删除key不支持模式匹配
                Set<String> keys = redisCacheProxy.keys(fullRegion + ":[ISO]:*");
                if (!keys.isEmpty())
                    redisCacheProxy.del(keys.toArray(new String[keys.size()]));
            } else {
                redisCacheProxy.del(byteFullRegion);
            }
        } catch (Exception e) {
            log.error("Error occured when clear data in redis2 cache", e);
            //throw new CacheException(e);
        }
    }

    public void destroy() throws CacheException {
        this.clear();
    }

    @Override
    public void put(Object key, Object value, Integer expireInSec) throws CacheException {
        if (key == null)
            return;
        if (value == null)
            evict(key);
        else {
            final String realKey = getRealKey(key);
            log.debug("PUT {}({}) {}s", fullRegion, realKey, expireInSec);
            try {
                if (storedInString)
                    redisCacheProxy.set(getRedisKeyNameBytes(realKey), SerializationUtils.serialize(value), expireInSec);
                else
                    redisCacheProxy.hset(byteFullRegion, realKey.getBytes(), SerializationUtils.serialize(value), expireInSec);
            } catch (Exception e) {
                log.error("Error occured when put expire data to redis2 cache", e);
                //throw new CacheException(e);
            }
        }
    }

    @Override
    public void update(Object key, Object value, Integer expireInSec) throws CacheException {
        put(key, value, expireInSec);
    }
}
